/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.dtub.mongo.update.startup;

import cn.ac.iie.di.ban.dtub.mongo.update.commons.RuntimeEnv;
import cn.ac.iie.di.ban.dtub.mongo.update.commons.RuntimeEnvException;
import cn.ac.iie.di.ban.dtub.mongo.update.exception.UpdateException;
import cn.ac.iie.di.ban.dtub.mongo.update.util.MongoHelper;
import cn.ac.iie.di.ban.dtub.mongo.update.worker.Receiver;
import cn.ac.iie.di.ban.dtub.mongo.update.util.SenderPool;
import cn.ac.iie.di.ban.dtub.mongo.update.worker.Sender;
import cn.ac.iie.di.ban.dtub.mongo.update.worker.SenderHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import java.util.HashMap;
import java.util.concurrent.ArrayBlockingQueue;
import org.apache.log4j.Logger;

/**
 *
 * @author zwlin
 */
public class DtubMongoUpdateStartup {

    public static boolean isTest = false;
    static private final Logger logger = Logger.getLogger(DtubMongoUpdateStartup.class.getName());
    private static SenderPool getSenderPool(String SendNamesrv, String SendTopic) throws REConnectionException {
        RESendSessionBuilder builder = (RESendSessionBuilder) new REConnection(SendNamesrv).getSendSessionBuilder(SendTopic);
        builder.addColumn("_id", REFieldType.String, true);
        builder.addColumn("_key", REFieldType.String, true);
        builder.addColumn("_spec", REFieldType.String, true);
        builder.addColumn("_ch", REFieldType.Long, true);
        builder.addColumn("nrd", REFieldType.Long, true);
        builder.addColumn("ut", REFieldType.Long, true);
        return new SenderPool(builder);
    }

    public static void main(String[] args) throws RuntimeEnvException, UpdateException, REConnectionException {
        RuntimeEnv.init();
        ArrayBlockingQueue<HashMap<String, Object>> queue = new ArrayBlockingQueue<>(10000);
        SenderPool[] pools = new SenderPool[2];
        pools[0] = getSenderPool(
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND1_NAMESRV),
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND1_TOPIC));
        pools[1] = getSenderPool(
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND2_NAMESRV),
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND2_TOPIC));
        MongoHelper[] helpers = new MongoHelper[2];
        helpers[0] = new MongoHelper(
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO1_HOSTS).split(";"),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO1_DB),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO1_TABLE),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO1_USER),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO1_PASSWD), "oldDB");
        helpers[1] = new MongoHelper(
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO2_HOSTS).split(";"),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO2_DB),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO2_TABLE),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO2_USER),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO2_PASSWD), "newDB");
        logger.info("haha");
        Receiver receiver = new Receiver(
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_RECEIVE_NAMESRV),
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_RECEIVE_TOPIC),
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_RECEIVE_GROUPNAME_PREFIX),
                pools,
                helpers,
                queue);
        receiver.start();

        Sender sender = new Sender(pools, helpers, queue);
        sender.start();
    }

}
